import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.observers.DisposableObserver;
import io.reactivex.schedulers.Schedulers;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import java.util.Arrays;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class Application {

    static CountDownLatch latch = new CountDownLatch(1);


    //注意，Observable是没有back pressure
    //注意，Flowable有back pressure
    public static void main(String[] args) throws InterruptedException {

        //testCallable();
        /*Observable<Long> o = testInterval();

        latch.await();*/
        /*testIntervalRange();
        latch.await();

        Observable o = Observable.empty();
        o.subscribe( System.out::println);*/

        //testTimer();
        //testDefer();
        testDisposableObserver();

    }

    private static void testAmb(){
        Observable<Integer> o1 = Observable.range(-10, 10)
                .skip(3)
                .repeat(2)
                .take(2);
        // o1.subscribe(System.out::println);

        Observable<Integer> o3 = Observable.range(2, 10);


        // o3.subscribe(System.out::println);

       /* Observable<String> o2 = Observable.just("hello")
                .repeat(3);
                o2.subscribe(System.out::println);*/




        Observable<Integer> ox = Observable.amb( Arrays.asList(o1,o3));
        ox.subscribe(System.out::println);
        System.out.println("------------------------");

        ox = Observable.concat(o1,o3);
        ox.subscribe(System.out::println);
    }
    private  static void testFromIterable(){
        Observable.fromIterable(Arrays.asList(
                "the",
                "quick",
                "brown",
                "fox",
                "jumped",
                "over",
                "the",
                "lazy",
                "dog"))
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println(s + "..........." + s);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }
    private static void testJust() {
        //创建 Observables
        //1. array
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
                .subscribe(System.out::println);

        //2. list
        Observable.just(Arrays.asList(
                "the",
                "quick",
                "brown",
                "fox",
                "jumped",
                "over",
                "the",
                "lazy",
                "dog"
        ))
                .subscribe(System.out::println);

        Observable.just("hangzhou", "ningbo")
                .subscribe(new SimpleObserver());
    }

    private static void testCallable(){
        Callable<String> c = () -> {
            Thread.sleep(3000);
            System.out.println("hello world");
            return "123456";
        };

        Observable o = Observable.fromCallable(c);

        o.subscribe( new Observer<String>(){
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                System.out.println("The result is: " +s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void testPublisher(){
        Publisher<String> publisher = new Publisher<String>() {

            @Override
            public void subscribe(Subscriber<? super String> s) {

            }
        };
    }

    private static Observable<Long> testInterval(){
        Observable<Long> c =
                Observable.interval(1, TimeUnit.SECONDS)
                .take(5);

        c.subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Long aLong) {
                System.out.println("ns....:" + aLong);
                if( aLong >19)
                    latch.countDown();
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
                latch.countDown();
            }
        });
        return c;
    }

    private static void testIntervalRange() {
        Observable<Long> c = Observable.intervalRange(0, 10, 0, 1, TimeUnit.SECONDS);

        c.subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Long aLong) {
                System.out.println("---------------*"+aLong);
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
                latch.countDown();
            }
        });
    }

    private static void testTimer(){
        Observable.timer(0, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Long aLong) {
                        System.out.println("+++++++++++++++++++:"+aLong);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    private static void testDefer() throws InterruptedException {
        Observable o = Observable.defer(() -> {
            Thread.sleep(2000);
            return Observable.just("un","deux","trois","quatre","cinq");
        });

        o.subscribeOn(Schedulers.io())
                .subscribe(new Observer() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Object o) {
                        System.out.println(o);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        latch.countDown();
                    }
                });
        latch.await();
    }

    private static void testDisposableObserver() throws InterruptedException {
        //Disposable d =
                Observable.intervalRange(0, 10, 0, 1, TimeUnit.SECONDS) //在worker线程上执行
                    //Observable.rangeLong(0,10)   //在主线程上执行
                        .subscribe(new DisposableObserver<Long>() {
                            @Override public void onStart() {
                                System.out.println("Start!");
                            }
                            @Override public void onNext(Long t) {
                                if (t == 3) {
                                    dispose();
                                    latch.countDown();
                                }
                                System.out.println(t);
                            }
                            @Override public void onError(Throwable t) {
                                t.printStackTrace();
                            }
                            @Override public void onComplete() {
                                System.out.println("Done!");
                            }
                        });
        // ...
        //d.dispose();
        System.out.println("main thread.");
        latch.await();
    }
}
